package global;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.jar.JarEntry;
import java.util.jar.JarOutputStream;

import org.apache.hadoop.mapred.JobConf;

public class GlobalJobConf {

	public static void setCommonProps(JobConf conf, String jobName) {
		conf.setJobName(jobName);

		conf.set("mapred.job.tracker", GlobalConsts.jobTrackerHost + ":" +
				GlobalConsts.jobTrackerPort);
		conf.set("fs.default.name", "hdfs://" + GlobalConsts.nameNodeHost + ":" +
				GlobalConsts.nameNodePort);
		conf.setJar(jobName + ".jar");

		System.setProperty("HADOOP_USER_NAME", GlobalConsts.hadoopUsername);

		generateJarFile(jobName);
	}

	public static void generateJarFile(String jarName) {
		try {
			String jarFileName = jarName + ".jar";
			File jarFile = new File(jarFileName);
			if ((jarFile.exists()) && (!jarFile.delete())) {
				throw new Exception();
			}
			JarOutputStream target = new JarOutputStream(new BufferedOutputStream(new FileOutputStream(jarFileName)));
			try {
				addToJar(target, new File("bin\\", "global\\"));
				addToJar(target, new File("bin\\", jarName + "\\"));
			}
			finally {
				target.close();
			}
		}
		catch (Exception ex) {
			System.out.println("Jar file could not be created.");
		}
	}

	private static void addToJar(JarOutputStream target, File source) throws IOException {
		BufferedInputStream in = null;
		try
		{
			if (source.isDirectory())
			{
				String name = source.getPath().replace("\\", "/");
				if (!name.isEmpty())
				{
					if (!name.endsWith("/"))
						name += "/";
					name.replace("bin/", "");
					JarEntry entry = new JarEntry(name);
					entry.setTime(source.lastModified());
					target.putNextEntry(entry);
					target.closeEntry();
				}
				for (File nestedFile: source.listFiles())
					addToJar(target, nestedFile);
				return;
			}

			JarEntry entry = new JarEntry(source.getPath().replace("\\", "/").replace("bin/", ""));
			entry.setTime(source.lastModified());
			target.putNextEntry(entry);
			in = new BufferedInputStream(new FileInputStream(source));

			byte[] buffer = new byte[1024];
			while (true)
			{
				int count = in.read(buffer);
				if (count == -1)
					break;
				target.write(buffer, 0, count);
			}
			target.closeEntry();
		}
		finally
		{
			if (in != null)
				in.close();
		}
	}
}
